package com.hsc.www.reSend;

import com.alibaba.fastjson.JSONObject;
import com.hsc.www.reSend.core.Config;
import com.hsc.www.reSend.core.DelayedMessageProperties;
import com.hsc.www.reSend.core.FreeBasicProperties;
import com.rabbitmq.client.AMQP;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;

import java.util.ArrayList;
import java.util.List;

public class RabbitTemplateExtend extends RabbitTemplate {

    public RabbitTemplateExtend() {

    }

    public RabbitTemplateExtend(ConnectionFactory connectionFactory) {
        super(connectionFactory);
    }

    private MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter();

    @Override
    public void send(String exchange, String routingKey, Message message, CorrelationData correlationData) throws AmqpException {

        DelayedAndTryMessage delayedAndTryMessage = DelayedMessageHelper.getDelayedMessage();
        if (delayedAndTryMessage != null) {
            try {
                AMQP.BasicProperties basicProperties = messagePropertiesConverter.fromMessageProperties(message.getMessageProperties(), "UTF-8");
                DelayedMessageProperties delayedMessageProperties = new DelayedMessageProperties();
                delayedMessageProperties.setBasicProperties(FreeBasicProperties.convert2FreeBasicProperties(basicProperties));
                delayedMessageProperties.setExchange(exchange);
                delayedMessageProperties.setRoutingKey(routingKey);
                delayedMessageProperties.setAlreadyTries(0);
                delayedMessageProperties.setTotalTries(delayedAndTryMessage.getTotalAttempts());
                delayedMessageProperties.setTryDelayedTimeList(delayedAndTryMessage.getDelayedTimeList());

                Integer delayTime = delayedAndTryMessage.getDelayTime();
                if (delayTime != null && delayTime > 0) {
                    exchange = "";
                    routingKey = Config.DELAYED_MESSAGE_REGISTRAR_QUEUE;
                    delayedMessageProperties.setDelayedMsgDelayedTime(delayedAndTryMessage.getDelayTime());
                    delayedMessageProperties.setDelayedMsgAlreadyDelay(false);
                }
                message.getMessageProperties().getHeaders().put(DelayedMessageProperties.X_ORIGINAL_QUEUE_INFO, JSONObject.toJSONString(delayedMessageProperties));
            } finally {
                DelayedMessageHelper.clearDelayedMessage();
            }
        }
        super.send(exchange, routingKey, message, correlationData);
    }


}
